package com.htdata.iiot.api.hbase.utils;

import com.htdata.iiot.api.hbase.pojo.DelMetric;
import com.htdata.iiot.api.tsdb.config.Configs;
import com.htdata.iiot.api.tsdb.utils.DBQuery;
import com.htdata.iiot.api.tsdb.utils.Tools;
import com.htdata.iiot.api.tsdb.utils.TsdbTools;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 使用hbase api删除tsdb数据
 * @author chensi
 */
public class DeleteTsdb {
    private static Logger LOGGER = Logger.getLogger(DeleteTsdb.class);
    public static List<byte[]> list = Collections.synchronizedList(new ArrayList<byte[]>());

    /**
     * 删除指定租户在某时间段的测点
     * @param tid
     * @param startT
     * @param enT
     * @param queryThread
     */
    public void delete(String tid,String startT,String enT,int queryThread,int delThread) throws Exception {
        LOGGER.info("正在查询myql...");
        long start = System.currentTimeMillis();
        //1.从mysql中查询租户下的所有dn
        List<String[]> metricL = getMetrciFromMysql(tid);
        long mysqlEndTime = System.currentTimeMillis() - start;
        LOGGER.info("从数据库获取dn耗时:"+mysqlEndTime+"ms");
        deleteMetricList(metricL,startT,enT,queryThread,delThread);
    }

    public void deleteMetricList(List<String[]> metricL,String startT,String enT,int queryThread,int delThread) throws Exception {
        int timeBase = Configs.TSDBConfig.METRIC_WIDTH + Configs.TSDBConfig.SALT_WIDTH;
        String[] times = Tools.splitDate(startT, enT);
        LOGGER.info("正在查询HBASE...");
        List<DelMetric> delL = new ArrayList<>();
        for(String[] s:metricL){
            delL.add(new DelMetric(s[0],null,s[1],s[2]));
            System.out.println(s[0]);
        }
        long startQ = System.currentTimeMillis();

        //2.查询rowkey
        List<DelMetric>[] lists = Tools.avgList(delL, queryThread);
        CountDownLatch cdl = new CountDownLatch(queryThread);
        for (List<DelMetric> list:lists){
            new Thread(new MutiQueryRowKey(list,cdl)).start();
        }
        //查询统计线程
        cdl.await();
        LOGGER.info("从HBASE中查询rowkey耗时:"+(System.currentTimeMillis()-startQ)+"ms|共需删除rowkey总数:"+list.size());
        LOGGER.info("正在删除...");

        //3.rowkey封装时间
        long startDel = System.currentTimeMillis();
        int days = times.length/24;
        int hours = times.length%24;
        LOGGER.info("共计需要删除:"+days+"天"+hours+"小时!");
        int index = 0;
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        List<byte[]> batchL = new ArrayList<>();
        for(int i=0;i<days;i++){
            long dayDelStart = System.currentTimeMillis();
            for(int j=0;j<24;j++){
                String time = times[index];
                byte[] rowKeyTime = TsdbTools.getRowKeyTime(sdf.parse(time).getTime());
                for(byte[] b:list){
                    b[timeBase] = rowKeyTime[0];
                    b[timeBase +1] = rowKeyTime[1];
                    b[timeBase +2] = rowKeyTime[2];
                    b[timeBase +3] = rowKeyTime[3];
                    byte[] tmp = b.clone();
//                    HbaseOperation.getOneRecord("tsdb",b);
                    batchL.add(tmp);
                }
                index++;
            }
            //执行删除
            List<byte[]>[] toDelL = Tools.avgList(batchL, delThread);
            CountDownLatch cdl2 = new CountDownLatch(delThread);
            for(List<byte[]> l:toDelL){
                new Thread(new MutiDel(l,cdl2)).start();
            }
            cdl2.await();
            LOGGER.info("删除第 "+(i+1)+" 天耗时: "+(System.currentTimeMillis() - dayDelStart)+" ms");
            for(byte[] b:batchL){
                System.out.println(Arrays.toString(b));
            }
            //empty list
            batchL = new ArrayList<>();
        }
        for(int i=0;i<hours;i++){
            String time = times[index];
            byte[] rowKeyTime = TsdbTools.getRowKeyTime(sdf.parse(time).getTime());
            for(byte[] b:list){
//                byte[] tmp = new byte[b.length];
                b[timeBase] = rowKeyTime[0];
                b[timeBase +1] = rowKeyTime[1];
                b[timeBase +2] = rowKeyTime[2];
                b[timeBase +3] = rowKeyTime[3];
                byte[] tmp = b.clone();
//                HbaseOperation.getOneRecord("tsdb",b);
                batchL.add(tmp);
            }
            index++;
        }
        //执行删除
        List<byte[]>[] toDelL = Tools.avgList(batchL, delThread);
        CountDownLatch cdl2 = new CountDownLatch(delThread);
        for(List<byte[]> l:toDelL){
            new Thread(new MutiDel(l,cdl2)).start();
        }
        cdl2.await();
        LOGGER.info("删除共计耗时: "+(System.currentTimeMillis() - startDel)+" ms");
        list = Collections.synchronizedList(new ArrayList<byte[]>());
    }

    /*public void delete(String tid,String startT,String enT,int queryThread,int delThread) throws Exception {
        String[] times = Tools.splitDate(startT, enT);
        LOGGER.info("正在查询myql...");
        long start = System.currentTimeMillis();

        //1.从mysql中查询租户下的所有dn
        List<String[]> metricL = getMetrciFromMysql(tid);
        long mysqlEndTime = System.currentTimeMillis() - start;
        LOGGER.info("从数据库获取dn耗时:"+mysqlEndTime+"ms");
        LOGGER.info("正在查询HBASE...");
        List<DelMetric> delL = new ArrayList<>();
        for(String[] s:metricL){
            delL.add(new DelMetric(s[0],times,s[1],s[2]));
        }
        long startQ = System.currentTimeMillis();

        //2.查询rowkey
        List<DelMetric>[] lists = Tools.avgList(delL, queryThread);
        CountDownLatch cdl = new CountDownLatch(queryThread);
        for (List<DelMetric> list:lists){
            new Thread(new MutiQueryRowKey(list,cdl)).start();
        }
        //查询统计线程
        cdl.await();
        LOGGER.info("从HBASE中查询rowkey耗时:"+(System.currentTimeMillis()-startQ)+"ms|共需删除rowkey总数:"+list.size());
        LOGGER.info("正在删除...");
        long startD = System.currentTimeMillis();

        //3.进行删除
        List<byte[]>[] toDelL = Tools.avgList(list, delThread);
        CountDownLatch cdl2 = new CountDownLatch(delThread);
        for(List<byte[]> l:toDelL){
            new Thread(new MutiDel(l,cdl2)).start();
        }
        cdl2.await();
        LOGGER.info("删除耗时:"+(System.currentTimeMillis() - startD)+"ms");
    }*/

    public List<String[]> getMetrciFromMysql(String tid){
        DBQuery dbQuery = new DBQuery();
        List<String[]> metricL = dbQuery.metircQueryWithTidDid(tid);
        return  metricL;
    }
}
class MutiDel implements Runnable{
    private Logger LOGGER = Logger.getLogger(MutiDel.class);
    private CountDownLatch cdl;
    private List<byte[]> list;
    public MutiDel(List<byte[]> list,CountDownLatch cdl){
        this.list = list;
        this.cdl = cdl;
    }
    @Override
    public void run() {
        try {
            LOGGER.info(Thread.currentThread().getName()+"当前线程需要删除rowkey数量为:"+list.size());
            HbaseOperation.batchDel(Configs.TSDBConfig.DATA_TABLE,list);
            cdl.countDown();
        } catch (IOException e) {
            LOGGER.error(e);
            cdl.countDown();
        }
    }
}
class MutiQueryRowKey implements Runnable{
    private Logger LOGGER = Logger.getLogger(MutiQueryRowKey.class);
    private CountDownLatch cdl;
    private List<DelMetric> list;
    public MutiQueryRowKey(List<DelMetric> list,CountDownLatch cdl){
        this.list = list;
        this.cdl = cdl;
    }
    @Override
    public void run() {
        try {
            List<byte[]> result = TsdbTools.getRowKeyFromMetricRefact(list);
            DeleteTsdb.list.addAll(result);
            cdl.countDown();
        } catch (Exception e) {
            LOGGER.error(e);
            cdl.countDown();
        }
    }
}
